package org.example.java_kafka;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;


public class MySimpleConsumer {


    private final static String TOPIC_NAME = "partition10";
    private final static String CONSUMER_GROUP_NAME = "testGroup";

    public static void main(String[] args) {
        //设置配置
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "192.168.2.135:9092,192.168.2.135:9093,192.168.2.135:9094");
        //消费分组名、key序列化、value序列化
        props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
//        // 是否自动提交offset，默认就是true 会丢消息
//        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
//        // 自动提交offset的间隔时间
//        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        //手动提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        //一次poll最大拉取消息的条数，可以根据消费速度的快慢来设置
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        //如果两次poll的时间如果超出了30s的时间间隔，kafka会认为其消费能力过弱，将其踢出消费组。将分区分配给其他消费者。-rebalance
//        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);
        //consumer给broker发送心跳的间隔时间
//        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);


        //1. 创建一个消费者客户端(设置配置文件)
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        //2. 消费者订阅主题列表
//        consumer.subscribe(Arrays.asList(TOPIC_NAME));
        //指定分区
        consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME,1)));
//        while (true) {
//            //3. poll() API是拉取消息的长轮询
//            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
//            for (ConsumerRecord<String, String> record : records) {
//                //4. 打印消息
//                System.out.printf("收到消息：partition = %d, offset = %d, key = %s, value = %s%n",
//                        record.partition(), record.offset(), record.key(), record.value());
//            }
//            //所有的消息已消费完
//            if (records.count() > 0) {//有消息
//                // 手动同步提交offset，当前线程会阻塞直到offset提交成功
//                // 一般使用同步提交，因为提交之后一般也没有什么逻辑代码了
//                consumer.commitSync();
//                // =======阻塞=== 提交成功
//            }
//
//                //异步提交
////                //所有的消息已消费完
////                if (records.count() > 0) {
////                    // 手动异步提交offset，当前线程提交offset不会阻塞，可以继续处理后面 的程序逻辑
////                    OffsetCommitCallback kafka集群回调方法处理提交后的业务（提交失败）
////                    consumer.commitAsync(new OffsetCommitCallback() {
////                        @Override
////                        public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
////                            if (exception != null) {
////                                System.err.println("Commit failed for " + offsets);
////                                System.err.println("Commit failed exception: " + exception.getStackTrace());
////                            }
////                        }
////                    });
////                }
//        }
        //指定时间消费
        List<PartitionInfo> topicPartitions = consumer.partitionsFor(TOPIC_NAME);
        //从1小时前开始消费
        long fetchDataTime = System.currentTimeMillis() - 1000 * 60 * 60;
        Map<TopicPartition, Long> map = new HashMap<>();
        for (PartitionInfo par : topicPartitions) {
            map.put(new TopicPartition(TOPIC_NAME, par.partition()), fetchDataTime);
        }
        Map<TopicPartition, OffsetAndTimestamp> parMap = consumer.offsetsForTimes(map);
        for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : parMap.entrySet()) {
            TopicPartition key = entry.getKey();
            OffsetAndTimestamp value = entry.getValue();
            if (key == null || value == null) {
                continue;
            }
            Long offset = value.offset();
            System.out.println("partition-" + key.partition() + "|offset-" + offset);
            System.out.println();
            //根据消费里的timestamp确定offset if (value != null) {
            consumer.assign(Arrays.asList(key));
            consumer.seek(key, offset);
        }
    }
}
